package org.firebug.spring.boot.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Future;

public class Client {
    public static void produce() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka.firebug.org:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        //
        ProducerRecord<String, String> record = new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
        try {
//            Future<RecordMetadata> future = producer.send(record);// 异步发送
//            RecordMetadata metadata = producer.send(record).get();// 同步发送
//            Future<RecordMetadata> future = producer.send(record, new Callback() { // 发送回调
//                @Override
//                public void onCompletion(RecordMetadata metadata, Exception e) {
//                    if(e != null) {
//                       e.printStackTrace();
//                    }
//                }
//            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void consume() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka.firebug.org:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("group.id", "G00");
        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
        consumer.subscribe(Collections.singletonList("CustomerCountry"));
    }
}
